package main

import (
	"encoding/json"
	"flag"
	"io/ioutil"
	"log"
	"net/http"
	"strings"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

const (
	namespace = "datanode"
)

var (
	listenAddress  = flag.String("web.listen-address", ":9075", "Address on which to expose metrics and web interface.")
	metricsPath    = flag.String("web.telemetry-path", "/metrics", "Path under which to expose metrics.")
	datanodeJmxUrl = flag.String("datanode.jmx.url", "http://127.0.0.1:50075/jmx", "Hadoop JMX URL.")
)

type Exporter struct {
	url                                        string
	BytesWritten                               prometheus.Gauge
	BytesRead                                  prometheus.Gauge
	BlocksWritten                              prometheus.Gauge
	BlocksRead                                 prometheus.Gauge
	BlocksReplicated                           prometheus.Gauge
	BlocksRemoved                              prometheus.Gauge
	BlocksVerified                             prometheus.Gauge
	BlockVerificationFailures                  prometheus.Gauge
	BlocksCached                               prometheus.Gauge
	BlocksUncached                             prometheus.Gauge
	RamDiskBlocksWrite                         prometheus.Gauge
	RamDiskBlocksWriteFallback                 prometheus.Gauge
	RamDiskBytesWrite                          prometheus.Gauge
	RamDiskBlocksReadHits                      prometheus.Gauge
	PacketAckRoundTripTimeNanosAvgTime         prometheus.Gauge
	SendDataPacketTransferNanosAvgTime         prometheus.Gauge
	SendDataPacketBlockedOnNetworkNanosAvgTime prometheus.Gauge
	FlushNanosNumOps                           prometheus.Gauge
	FlushNanosAvgTime                          prometheus.Gauge
}

func NewExporter(url string) *Exporter {
	return &Exporter{
		url: url,
		BytesWritten: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BytesWritten",
			Help:      "Total number of bytes written to DataNode",
		}),
		BytesRead: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BytesRead",
			Help:      "Total number of bytes read from DataNode",
		}),
		BlocksWritten: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksWritten",
			Help:      "Total number of blocks written to DataNode",
		}),
		BlocksRead: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksRead",
			Help:      "Total number of blocks read from DataNodeg",
		}),
		BlocksReplicated: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksReplicated",
			Help:      "Total number of blocks replicated",
		}),
		BlocksRemoved: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksRemoved",
			Help:      "Total number of blocks removed",
		}),
		BlocksVerified: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksVerified",
			Help:      "Total number of blocks verified",
		}),
		BlockVerificationFailures: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlockVerificationFailures",
			Help:      "Total number of verifications failures",
		}),
		BlocksCached: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksCached",
			Help:      "Total number of blocks cached",
		}),
		BlocksUncached: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "BlocksUncached",
			Help:      "Total number of blocks uncached",
		}),
		RamDiskBlocksWrite: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "RamDiskBlocksWrite",
			Help:      "Total number of blocks written to memory",
		}),
		RamDiskBlocksWriteFallback: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "RamDiskBlocksWriteFallback",
			Help:      "Total number of blocks written to memory but not satisfied",
		}),
		RamDiskBytesWrite: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "RamDiskBytesWrite",
			Help:      "Total number of bytes written to memory",
		}),
		RamDiskBlocksReadHits: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "RamDiskBlocksReadHits",
			Help:      "Total number of times a block in memory was read",
		}),
		PacketAckRoundTripTimeNanosAvgTime: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "PacketAckRoundTripTimeNanosAvgTime",
			Help:      "Average time from ack send to receive minus the downstream ack time in nanoseconds",
		}),
		SendDataPacketTransferNanosAvgTime: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "SendDataPacketTransferNanosAvgTime",
			Help:      "Average transfer time of sending packets in nanoseconds",
		}),
		SendDataPacketBlockedOnNetworkNanosAvgTime: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "SendDataPacketBlockedOnNetworkNanosAvgTime",
			Help:      "hAverage waiting time of sending packets in nanoseconds",
		}),
		FlushNanosNumOps: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "FlushNanosNumOps",
			Help:      "Total number of flushes",
		}),
		FlushNanosAvgTime: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: namespace,
			Name:      "FlushNanosAvgTime",
			Help:      "Average flush time in nanoseconds",
		}),
	}
}

// Describe implements the prometheus.Collector interface.
func (e *Exporter) Describe(ch chan<- *prometheus.Desc) {
	e.BytesWritten.Describe(ch)
	e.BytesRead.Describe(ch)
	e.BlocksWritten.Describe(ch)
	e.BlocksRead.Describe(ch)
	e.BlocksReplicated.Describe(ch)
	e.BlocksRemoved.Describe(ch)
	e.BlocksVerified.Describe(ch)
	e.BlockVerificationFailures.Describe(ch)
	e.BlocksCached.Describe(ch)
	e.BlocksUncached.Describe(ch)
	e.RamDiskBlocksWrite.Describe(ch)
	e.RamDiskBlocksWriteFallback.Describe(ch)
	e.RamDiskBytesWrite.Describe(ch)
	e.RamDiskBlocksReadHits.Describe(ch)
	e.PacketAckRoundTripTimeNanosAvgTime.Describe(ch)
	e.SendDataPacketTransferNanosAvgTime.Describe(ch)
	e.SendDataPacketBlockedOnNetworkNanosAvgTime.Describe(ch)
	e.FlushNanosNumOps.Describe(ch)
	e.FlushNanosAvgTime.Describe(ch)
}

// Collect implements the prometheus.Collector interface.
func (e *Exporter) Collect(ch chan<- prometheus.Metric) {
	resp, err := http.Get(e.url)
	if err != nil {
		log.Print(err)
	}
	defer resp.Body.Close()
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Print(err)
	}
	var f interface{}
	err = json.Unmarshal(data, &f)
	if err != nil {
		log.Print(err)
	}
	// {"beans":[{"name":"Hadoop:service=NameNode,name=FSNamesystem", ...}, {"name":"java.lang:type=MemoryPool,name=Code Cache", ...}, ...]}
	m := f.(map[string]interface{})
	// [{"name":"Hadoop:service=NameNode,name=FSNamesystem", ...}, {"name":"java.lang:type=MemoryPool,name=Code Cache", ...}, ...]
	var nameList = m["beans"].([]interface{})
	for _, nameData := range nameList {
		nameDataMap := nameData.(map[string]interface{})
		/*
						{
			    "name" : "Hadoop:service=DataNode,name=DataNodeActivity-quickstart.cloudera-50010",
			    "modelerType" : "DataNodeActivity-quickstart.cloudera-50010",
			    "tag.SessionId" : null,
			    "tag.Context" : "dfs",
			    "tag.Hostname" : "quickstart.cloudera",
			    "BytesWritten" : 16662,
			    "TotalWriteTime" : 3,
			    "BytesRead" : 6895,
			    "TotalReadTime" : 22,
			    "BlocksWritten" : 107,
			    "BlocksRead" : 106,
			    "BlocksReplicated" : 0,
			    "BlocksRemoved" : 108,
			    "BlocksVerified" : 0,
			    "BlockVerificationFailures" : 0,
			    "BlocksCached" : 0,
			    "BlocksUncached" : 0,
			    "ReadsFromLocalClient" : 106,
			    "ReadsFromRemoteClient" : 0,
			    "WritesFromLocalClient" : 107,
			    "WritesFromRemoteClient" : 0,
			    "BlocksGetLocalPathInfo" : 6,
			    "RemoteBytesRead" : 0,
			    "RemoteBytesWritten" : 0,
			    "RamDiskBlocksWrite" : 0,
			    "RamDiskBlocksWriteFallback" : 0,
			    "RamDiskBytesWrite" : 0,
			    "RamDiskBlocksReadHits" : 0,
			    "RamDiskBlocksEvicted" : 0,
			    "RamDiskBlocksEvictedWithoutRead" : 0,
			    "RamDiskBlocksEvictionWindowMsNumOps" : 0,
			    "RamDiskBlocksEvictionWindowMsAvgTime" : 0.0,
			    "RamDiskBlocksLazyPersisted" : 0,
			    "RamDiskBlocksDeletedBeforeLazyPersisted" : 0,
			    "RamDiskBytesLazyPersisted" : 0,
			    "RamDiskBlocksLazyPersistWindowMsNumOps" : 0,
			    "RamDiskBlocksLazyPersistWindowMsAvgTime" : 0.0,
			    "FsyncCount" : 12,
			    "VolumeFailures" : 0,
			    "DatanodeNetworkErrors" : 0,
			    "ReadBlockOpNumOps" : 106,
			    "ReadBlockOpAvgTime" : 3.0,
			    "WriteBlockOpNumOps" : 107,
			    "WriteBlockOpAvgTime" : 9.0,
			    "BlockChecksumOpNumOps" : 0,
			    "BlockChecksumOpAvgTime" : 0.0,
			    "CopyBlockOpNumOps" : 0,
			    "CopyBlockOpAvgTime" : 0.0,
			    "ReplaceBlockOpNumOps" : 0,
			    "ReplaceBlockOpAvgTime" : 0.0,
			    "HeartbeatsNumOps" : 2040,
			    "HeartbeatsAvgTime" : 1.2666666666666666,
			    "BlockReportsNumOps" : 2,
			    "BlockReportsAvgTime" : 15.0,
			    "IncrementalBlockReportsNumOps" : 111,
			    "IncrementalBlockReportsAvgTime" : 1.0,
			    "CacheReportsNumOps" : 514,
			    "CacheReportsAvgTime" : 1.0,
			    "PacketAckRoundTripTimeNanosNumOps" : 0,
			    "PacketAckRoundTripTimeNanosAvgTime" : 0.0,
			    "FlushNanosNumOps" : 235,
			    "FlushNanosAvgTime" : 121314.5,
			    "FsyncNanosNumOps" : 24,
			    "FsyncNanosAvgTime" : 872898.4583333334,
			    "SendDataPacketBlockedOnNetworkNanosNumOps" : 212,
			    "SendDataPacketBlockedOnNetworkNanosAvgTime" : 28318.5,
			    "SendDataPacketTransferNanosNumOps" : 212,
			    "SendDataPacketTransferNanosAvgTime" : 11016.5
			  }
		*/
		if strings.HasPrefix(nameDataMap["name"].(string), "Hadoop:service=DataNode,name=DataNodeActivity") {
			e.BytesWritten.Set(nameDataMap["BytesWritten"].(float64))
			e.BytesRead.Set(nameDataMap["BytesRead"].(float64))
			e.BlocksWritten.Set(nameDataMap["BlocksWritten"].(float64))
			e.BlocksRead.Set(nameDataMap["BlocksRead"].(float64))
			e.BlocksReplicated.Set(nameDataMap["BlocksReplicated"].(float64))
			e.BlocksRemoved.Set(nameDataMap["BlocksRemoved"].(float64))
			e.BlocksVerified.Set(nameDataMap["BlocksVerified"].(float64))
			e.BlockVerificationFailures.Set(nameDataMap["BlockVerificationFailures"].(float64))
			e.BlocksCached.Set(nameDataMap["BlocksCached"].(float64))
			e.BlocksUncached.Set(nameDataMap["BlocksUncached"].(float64))
			e.RamDiskBlocksWrite.Set(nameDataMap["RamDiskBlocksWrite"].(float64))
			e.RamDiskBlocksWriteFallback.Set(nameDataMap["RamDiskBytesWrite"].(float64))
			e.RamDiskBlocksReadHits.Set(nameDataMap["RamDiskBlocksReadHits"].(float64))
			e.PacketAckRoundTripTimeNanosAvgTime.Set(nameDataMap["PacketAckRoundTripTimeNanosAvgTime"].(float64))
			e.SendDataPacketTransferNanosAvgTime.Set(nameDataMap["SendDataPacketTransferNanosAvgTime"].(float64))
			e.SendDataPacketBlockedOnNetworkNanosAvgTime.Set(nameDataMap["SendDataPacketBlockedOnNetworkNanosAvgTime"].(float64))
			e.FlushNanosNumOps.Set(nameDataMap["FlushNanosNumOps"].(float64))
			e.FlushNanosAvgTime.Set(nameDataMap["FlushNanosAvgTime"].(float64))
		}
	}
	e.BytesWritten.Collect(ch)
	e.BytesRead.Collect(ch)
	e.BlocksWritten.Collect(ch)
	e.BlocksRead.Collect(ch)
	e.BlocksReplicated.Collect(ch)
	e.BlocksRemoved.Collect(ch)
	e.BlocksVerified.Collect(ch)
	e.BlockVerificationFailures.Collect(ch)
	e.BlocksCached.Collect(ch)
	e.BlocksUncached.Collect(ch)
	e.RamDiskBlocksWrite.Collect(ch)
	e.RamDiskBlocksWriteFallback.Collect(ch)
	e.RamDiskBytesWrite.Collect(ch)
	e.RamDiskBlocksReadHits.Collect(ch)
	e.PacketAckRoundTripTimeNanosAvgTime.Collect(ch)
	e.SendDataPacketTransferNanosAvgTime.Collect(ch)
	e.SendDataPacketBlockedOnNetworkNanosAvgTime.Collect(ch)
	e.FlushNanosNumOps.Collect(ch)
	e.FlushNanosAvgTime.Collect(ch)

}

func main() {
	flag.Parse()

	exporter := NewExporter(*datanodeJmxUrl)
	prometheus.MustRegister(exporter)

	log.Printf("Starting Server: %s", *listenAddress)
	http.Handle(*metricsPath, promhttp.Handler())
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Write([]byte(`<html>
		<head><title>DataNode Exporter</title></head>
		<body>
		<h1>DataNode Exporter</h1>
		<p><a href="` + *metricsPath + `">Metrics</a></p>
		</body>
		</html>`))
	})
	err := http.ListenAndServe(*listenAddress, nil)
	if err != nil {
		log.Print(err)
	}
}
